/**
 * Copyright (C) 2010, 2011 Neofonie GmbH
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package eu.dicodeproject.twitterstream.sink;

import java.io.IOException;
import java.net.URL;
import java.util.Map;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.HTablePool;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import eu.dicodeproject.analysis.hbase.TweetCols;

import eu.dicodeproject.analysis.util.Language;
import eu.dicodeproject.twitterstream.vectorize.Vectorizer;
import twitter4j.GeoLocation;
import twitter4j.HashtagEntity;
import twitter4j.Place;
import twitter4j.Status;
import twitter4j.URLEntity;
import twitter4j.User;
import twitter4j.UserMentionEntity;

/**
 * Class that stores tweets to HBase.
 */
public class HBaseTweetSink implements TweetSink {
  /** Logger for this class. */
  private static final Logger log = LoggerFactory.getLogger(HBaseTweetSink.class);

  /** Column families. */
  private byte[] tweetFamily;
  private byte[] vectorFamily;
  private byte[] userFamily;
  /** Tweet table name. */
  private String tweetTableName;
  /** User table name. */
  private String userTableName;
  /** Zookeeper Quorum **/
  private String zookeeperQuorum = "localhost";
  /** Zookeeper Port **/
  private int zookeeperPort = 2181;
  /** HBase configuration. */
  private Configuration configuration;

  private HTablePool hTablePool;

  /** Class that creates vectors out of text. */
  private Vectorizer vectorizer;

  /** Tweet counter for logging **/
  private static long counter;

  /** DateFormat used to parse tweet's creation date */
  // private final DateFormat dateFormat = new
  // SimpleDateFormat("EEE, dd MMM yyyy HH:mm:ss +SSS");

  /**
   * If a {@link Configuration} was set, that instance will be used. Otherwise
   * {@link #init()} creates a new one using the values given by
   * {@link #setZookeeperPort(int)} and {@link #setZookeeperQuorum(String)}.
   * 
   * @throws IOException
   */
  @PostConstruct
  public void init() throws IOException {
    if (this.configuration == null) {
      this.configuration = new Configuration();
      this.configuration.clear();
      this.configuration.set("hbase.zookeeper.quorum", this.zookeeperQuorum);
      this.configuration.setInt("hbase.zookeeper.property.clientPort", this.zookeeperPort);
    }
    HBaseAdmin admin = new HBaseAdmin(this.configuration);

    initTable(this.tweetTableName, admin, this.tweetFamily, this.vectorFamily);
    initTable(this.userTableName, admin, this.userFamily);

    this.hTablePool = new HTablePool(this.configuration, 100);
  }

  @PreDestroy
  public void shutdown() {
    this.hTablePool.closeTablePool(this.userTableName);
    this.hTablePool.closeTablePool(this.tweetTableName);
  }

  private void initTable(String tablename, HBaseAdmin admin, byte[]... families) throws IOException {
    if (!this.checkTable(tablename, admin)) {
      HTableDescriptor desc = new HTableDescriptor(tablename);
      for (byte[] family : families) {
        HColumnDescriptor colFam = new HColumnDescriptor(family);
        desc.addFamily(colFam);

        if (log.isInfoEnabled()) {
          StringBuilder message = new StringBuilder("Creating table ");
          message.append(tablename);
          message.append(" with column family " + Bytes.toString(family) + " as non were found.");
          log.info(message.toString());
        }
      }
      admin.createTable(desc);
    }
  }

  private boolean checkTable(final String tablename, final HBaseAdmin admin) throws IOException {
    HTableDescriptor[] descs = admin.listTables();
    for (HTableDescriptor desc : descs) {
      if (tablename.equals(desc.getNameAsString())) {
        return true;
      }
    }
    return false;
  }

  private void add(final byte[] family, final byte[] col, final String value, Put targetPut) {
    if (value != null) {
      targetPut.add(family, col, Bytes.toBytes(value));
    }
  }

  private void add(final byte[] family, final byte[] col, final long value, Put targetPut) {
    targetPut.add(family, col, Bytes.toBytes(value));
  }

  private void add(final byte[] family, final byte[] col, final int value, Put targetPut) {
    targetPut.add(family, col, Bytes.toBytes(value));
  }

  private void add(final byte[] family, final byte[] col, final double value, Put targetPut) {
    targetPut.add(family, col, Bytes.toBytes(value));
  }

  /** {@inheritDoc} */
  @Override
  public void store(final Status status) throws IOException {
    HTableInterface tweetTable = this.hTablePool.getTable(this.tweetTableName);
    HTableInterface userTable = this.hTablePool.getTable(this.userTableName);

    try {
      storeTweetData(status, tweetTable);
      storeUserData(status, userTable);

      // We want to track the number of stored tweets since restart
      if (++counter % 100000 == 0) {
        log.info(counter + " since restart / 100,000 tweets stored");
      }

    } catch (Exception e) {
      // we expect there to be parsing trouble and do not want to throw away the
      // whole batch of tweets in that case.
      log.warn(
          "Error processing tweet: "
              + e.getMessage()
              + " (Ignoring and continuing processing - be alarmed if that happens often for your desired definition of often.) ",
          e);
      log.warn("Tweet that let to the exception: " + status);
    } finally {
      this.hTablePool.putTable(userTable);
      this.hTablePool.putTable(tweetTable);
    }
  }

  /**
   * Store all user related data in the userTable
   * 
   * @param status
   * @param userTable
   * @throws IOException
   */
  private void storeUserData(final Status status, HTableInterface userTable) throws IOException {
    Put putUser = null;

    if (status.getUser() != null) {
      User user = status.getUser();

      // create put using twitter's user id as row key.
      // twitter's user ids from search api != user ids from twitter's rest
      // api...
      putUser = new Put(Bytes.toBytes(user.getId()));

      if (user.getProfileImageURL() != null) {
        add(userFamily, TweetCols.IMAGE_URL.bytes(), user.getProfileImageURL().toString(), putUser);
      }
      add(userFamily, TweetCols.LANG.bytes(), user.getLang(), putUser);

      if (user.getCreatedAt() != null) {
        add(userFamily, TweetCols.USER_CREATED_AT.bytes(), user.getCreatedAt().getTime(), putUser); // long
      }
      add(userFamily, TweetCols.USER_DESCRIPTION.bytes(), user.getDescription(), putUser);
      add(userFamily, TweetCols.USER_FAVOURITES_COUNT.bytes(), user.getFavouritesCount(), putUser); // int
      add(userFamily, TweetCols.USER_FOLLOWERS_COUNT.bytes(), user.getFollowersCount(), putUser); // int
      add(userFamily, TweetCols.USER_FRIENDS_COUNT.bytes(), user.getFriendsCount(), putUser); // int
      if (user.getListedCount() > 0) {
        add(userFamily, TweetCols.USER_LISTED_COUNT.bytes(), user.getListedCount(), putUser); // int
      }
      add(userFamily, TweetCols.USER_LOCATION.bytes(), user.getLocation(), putUser);
      add(userFamily, TweetCols.USER_NAME.bytes(), user.getName(), putUser);
      add(userFamily, TweetCols.USER_PROFILE_BACKGROUND_COLOR.bytes(), user.getProfileBackgroundColor(),
          putUser);
      add(userFamily, TweetCols.USER_PROFILE_BACKGROUND_IMAGE_URL.bytes(),
          user.getProfileBackgroundImageUrl(), putUser);

      if (user.getProfileImageURL() != null) {
        add(userFamily, TweetCols.USER_PROFILE_IMAGE_URL.bytes(), user.getProfileImageURL().toString(),
            putUser);
      }
      add(userFamily, TweetCols.USER_PROFILE_LINK_COLOR.bytes(), user.getProfileLinkColor(), putUser);
      add(userFamily, TweetCols.USER_PROFILE_SIDEBAR_BORDER_COLOR.bytes(),
          user.getProfileSidebarBorderColor(), putUser);
      add(userFamily, TweetCols.USER_PROFILE_SIDEBAR_FILL_COLOR.bytes(), user.getProfileSidebarFillColor(),
          putUser);
      add(userFamily, TweetCols.USER_PROFILE_TEXT_COLOR.bytes(), user.getProfileTextColor(), putUser);
      add(userFamily, TweetCols.USER_SCREEN_NAME.bytes(), user.getScreenName(), putUser);
      add(userFamily, TweetCols.USER_STATUSES_COUNT.bytes(), user.getStatusesCount(), putUser); // int
      add(userFamily, TweetCols.USER_TIMEZONE.bytes(), user.getTimeZone(), putUser);
      if (user.getURL() != null) {
        add(userFamily, TweetCols.USER_URL.bytes(), user.getURL().toString(), putUser);
      }
      add(userFamily, TweetCols.USER_UTC_OFFSET.bytes(), user.getUtcOffset(), putUser); // int
    }
    if (putUser != null) {
      userTable.put(putUser);
    }
  }

  /**
   * Store the tweet data in the tweetTable
   * 
   * @param status
   * @param tweetTable
   * @throws IOException
   */
  private void storeTweetData(final Status status, HTableInterface tweetTable) throws IOException {
    // we re-use the original tweet id from twitter as uuid
    // for more info on their generation implementation see
    // https://github.com/twitter/snowflake
    Put putTweet = new Put(Bytes.toBytes(status.getId()));

    String language = "";
    if (status.getUser() != null) {
      User user = status.getUser();
      add(tweetFamily, TweetCols.FROM.bytes(), user.getName(), putTweet);
      add(tweetFamily, TweetCols.FROM_ID.bytes(), user.getId(), putTweet); // long
      language = user.getLang();
      add(tweetFamily, TweetCols.LANG.bytes(), language, putTweet);
    }

    // Twitter returns a localized date: Fri Aug 19 16:33:50 CEST 2011
    if (status.getCreatedAt() != null) {
      add(tweetFamily, TweetCols.CREATION_DATE.bytes(), status.getCreatedAt().getTime(), putTweet);
    }

    if (status.getInReplyToStatusId() > 0) {
      add(tweetFamily, TweetCols.IN_REPLY_TO_STATUS_ID.bytes(), "" + status.getInReplyToStatusId(), putTweet);
    }

    add(tweetFamily, TweetCols.SOURCE.bytes(), status.getSource(), putTweet);

    String tweetText = status.getText();
    add(tweetFamily, TweetCols.TEXT.bytes(), tweetText, putTweet);

    // add term vector values in row with d:term1 val, d:term2 val, etc.
    // choice of analyzer based of language specified in user profile (Tweet
    // text language recognition?)
    if (vectorizer != null) {

      vectorizer.setLanguage(Language.fromCode(language));

      for (Map.Entry<String, Integer> entry : vectorizer.createVector(tweetText).entrySet()) {
        add(vectorFamily, Bytes.toBytes(entry.getKey()), entry.getValue(), putTweet);
      }
    }

    if (status.getInReplyToUserId() > 0) {
      add(tweetFamily, TweetCols.TO.bytes(), "" + status.getInReplyToUserId(), putTweet);
    }

    addHashtags(status, putTweet);
    addLocationData(status, putTweet);
    addFirstUrl(status, putTweet);
    addUsersMentioned(status, putTweet);

    if (status.getRetweetCount() > 0) {
      add(tweetFamily, TweetCols.RETWEET_COUNT.bytes(), status.getRetweetCount(), putTweet); // long
    }
    // will this ever happen?
    if (status.getRetweetedStatus() != null) {
      this.store(status.getRetweetedStatus());
    }

    tweetTable.put(putTweet);
  }

  /**
   * Append userIds and userNamesn mentioned to a string separated by hash marks
   * 
   * @param status
   * @param putTweet
   */
  private void addUsersMentioned(final Status status, Put putTweet) {
    UserMentionEntity[] userMentionEntities = status.getUserMentionEntities();
    if (userMentionEntities != null && userMentionEntities.length > 0) {
      StringBuilder userIds = new StringBuilder();
      StringBuilder userNames = new StringBuilder();
      for (UserMentionEntity entity : userMentionEntities) {
        userIds.append('#');
        userIds.append(entity.getId());
        userNames.append('#');
        userNames.append(entity.getName());
      }
      add(tweetFamily, TweetCols.USER_IDS_MENTIONED.bytes(), userIds.toString(), putTweet);
      add(tweetFamily, TweetCols.USER_NAMES_MENTIONED.bytes(), userNames.toString(), putTweet);
    }
  }

  /**
   * Append all hashtags to a string separated by hash marks
   * 
   * @param status
   * @param putTweet
   */
  private void addHashtags(final Status status, Put putTweet) {
    HashtagEntity[] hashtagEntities = status.getHashtagEntities();
    if (hashtagEntities != null && hashtagEntities.length > 0) {
      StringBuilder hashTagsBuilder = new StringBuilder();
      for (HashtagEntity tag : hashtagEntities) {
        hashTagsBuilder.append('#');
        hashTagsBuilder.append(tag.getText());
      }
      add(tweetFamily, TweetCols.HASHTAGS.bytes(), hashTagsBuilder.toString(), putTweet);
    }
  }

  /**
   * We only add the first Url (most tweets only contain a single url)
   * 
   * @param status
   * @param putTweet
   */
  private void addFirstUrl(final Status status, Put putTweet) {
    URLEntity[] urlEntities = status.getURLEntities();
    URL url = null;

    if (urlEntities != null && urlEntities.length > 0) {
      URLEntity entity = urlEntities[0];
      // we prefer the expanded url ...
      url = entity.getExpandedURL();
      if (url == null) {
        url = entity.getURL();
      }
      add(tweetFamily, TweetCols.URL.bytes(), url.toString(), putTweet);
    }
  }

  /**
   * Add geo related data.
   * 
   * @param status
   * @param putTweet
   */
  private void addLocationData(final Status status, Put putTweet) {

    boolean hasGeodata = false;
    double placeLatitude = 0;
    double placeLongitude = 0;

    // add geoLocation JSON
    if (status.getGeoLocation() != null) {
      add(tweetFamily, TweetCols.GEO.bytes(), status.getGeoLocation().toString(), putTweet);
    }

    if (status.getPlace() != null) {
      Place place = status.getPlace();

      // add place meta data (there's more data available which we could store)
      add(tweetFamily, TweetCols.PLACE_COUNTRY_CODE.bytes(), place.getCountryCode(), putTweet);
      add(tweetFamily, TweetCols.PLACE_FULL_NAME.bytes(), place.getFullName(), putTweet);
      add(tweetFamily, TweetCols.PLACE_TYPE.bytes(), place.getPlaceType(), putTweet);

      // 1: check if we have point coordinates first
      if (place.getGeometryType() != null && place.getGeometryType().equals("Point")) {
        GeoLocation[][] coordinates = place.getGeometryCoordinates();
        placeLatitude = coordinates[0][0].getLatitude();
        placeLongitude = coordinates[0][0].getLongitude();
        hasGeodata = true;
      }

      // 2. use location geo coordinates instead
      if (!hasGeodata && status.getGeoLocation() != null) {
        placeLatitude = status.getGeoLocation().getLatitude();
        placeLongitude = status.getGeoLocation().getLongitude();
        hasGeodata = true;
      }

      // 3. still nothing found? We take the first poligon coordinate
      // TODO: rather use the center of the polygon?
      if (!hasGeodata && place.getBoundingBoxType() != null && place.getBoundingBoxType().equals("Polygon")) {
        GeoLocation[][] coordinates = place.getBoundingBoxCoordinates();
        placeLatitude = coordinates[0][0].getLatitude();
        placeLongitude = coordinates[0][0].getLongitude();

        hasGeodata = true;
      }
    }
    if (hasGeodata) {
      add(tweetFamily, TweetCols.PLACE_LATITUDE.bytes(), placeLatitude, putTweet); // double
      add(tweetFamily, TweetCols.PLACE_LONGITUDE.bytes(), placeLongitude, putTweet); // double
    }
  }

  /**
   * @param tweetFamily
   *          the tweetFamily to set
   */
  public void setTweetFamily(String tweetFamily) {
    this.tweetFamily = tweetFamily.getBytes();
  }

  /**
   * @param tweetVectorsFamily
   *          the tweetVectorsFamily to set
   */
  public void setTweetVectorsFamily(String tweetVectorsFamily) {
    this.vectorFamily = tweetVectorsFamily.getBytes();
  }

  /**
   * @param userFamily
   *          the userFamily to set
   */
  public void setUserFamily(String userFamily) {
    this.userFamily = userFamily.getBytes();
  }

  /**
   * @param tweetTableName
   *          the tweetTableName to set
   */
  public void setTweetTableName(String tweetTableName) {
    this.tweetTableName = tweetTableName.trim();
  }

  /**
   * @param userTableName
   *          the userTableName to set
   */
  public void setUserTableName(String userTableName) {
    this.userTableName = userTableName.trim();
  }

  /**
   * @param zookeeperQuorum
   *          the zookeeperQuorum to set
   */
  public void setZookeeperQuorum(String zookeeperQuorum) {
    this.zookeeperQuorum = zookeeperQuorum;
  }

  /**
   * @param zookeeperPort
   *          the zookeeperPort to set
   */
  public void setZookeeperPort(int zookeeperPort) {
    this.zookeeperPort = zookeeperPort;
  }

  /**
   * @param vectorizer
   *          the vectorizer to set
   */
  public void setVectorizer(Vectorizer vectorizer) {
    this.vectorizer = vectorizer;
  }

  /**
   * @param configuration
   *          the configuration to set
   */
  void setHBaseConfiguration(Configuration configuration) {
    this.configuration = configuration;
  }

}
